package com.echo.code.thread.mode.worker;

import java.util.Arrays;
import java.util.logging.Logger;
import java.util.stream.IntStream;

/**
 * Channel Created by baiyap on 2018/9/24.
 */
public class Channel {

  private static final Logger LOGGER = Logger.getLogger(Channel.class.getName());

  private final static int MAX_WORK_LENGTH = 100;
  private final Work[] works;
  private final Worker[] workers;
  private int tail;
  private int head;
  private int total;

  public Channel(int workerSize) {
    if (workerSize < 1) {
      throw new IllegalArgumentException("illegal argument");
    }
    this.works = new ConcreteWork[MAX_WORK_LENGTH];
    this.workers = new Worker[workerSize];
    // --------------initialization--------------
    IntStream.range(0, workerSize).forEach(i -> workers[i] = new Worker(this));
    Arrays.stream(workers).forEach(Thread::start);
    // ------------------------------------------
  }

  // offer work
  public void offerWork(Work work) throws InterruptedException {
    synchronized (this) {
      // when exceed max
      while (total >= works.length) {
        LOGGER.info(getCurrentThreadName() + " suspend, waiting offer...");
        this.wait();
      }
      // add work
      works[tail] = work;
      LOGGER.info(getCurrentThreadName() + " add work at position-" + tail);
      tail = (tail + 1) % works.length;
      total++;
      this.notifyAll();
    }
  }

  // take work
  public Work takeWork() throws InterruptedException {
    synchronized (this) {
      // when not exceed min
      while (total <= 0) {
        LOGGER.info(getCurrentThreadName() + " suspend, waiting take...");
        this.wait();
      }
      // remove work
      Work work = works[head];
      LOGGER.info(getCurrentThreadName() + " remove work at position-" + head);
      head = (head + 1) % works.length;
      total--;
      this.notifyAll();
      return work;
    }
  }

  private String getCurrentThreadName() {
    return Thread.currentThread().getName();
  }
}

